package com.lonni.rpc.core.socket.client;

import com.lonni.rpc.core.socket.SocketUtil;
import com.lonni.rpc.core.socket.codec.RpcProtocolDecoder;
import com.lonni.rpc.core.socket.codec.RpcProtocolEncoder;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LoggingHandler;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import java.io.Serializable;
import java.net.InetSocketAddress;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ThreadFactory;

/**
 * rpc客户端
 *
 * @author: Lonni
 * @date: 2024/7/9 0009 10:01
 */
@Slf4j
public class RpcNettyClient implements Serializable {

    public static RpcNettyClient client = new RpcNettyClient();

    private final Bootstrap bootstrap;

    private final EventLoopGroup workGroup;
    private ChannelFuture channelFuture;


    ThreadFactory workFactory = r -> {
        Thread thread = new Thread(r);
        thread.setName("rpc-netty-work-client-");
        return thread;
    };

    private RpcNettyClient() {
        bootstrap = new Bootstrap();
        workGroup = Epoll.isAvailable() ? new EpollEventLoopGroup(workFactory) : new NioEventLoopGroup(workFactory);
        bootstrap.group(workGroup)
                .channel(Epoll.isAvailable() ? EpollSocketChannel.class : NioSocketChannel.class)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .handler(new LoggingHandler())
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel channel) throws Exception {
                        ChannelPipeline pipeline = channel.pipeline();
                        pipeline.addLast(new RpcProtocolEncoder())
                                .addLast(new RpcProtocolDecoder())
                                .addLast(new RpcClientHandler());
                    }
                });
    }


    /**
     * 链接服务
     *
     * @return
     */
    public Channel connected(InetSocketAddress address) throws ExecutionException, InterruptedException {

        CompletableFuture<Channel> channelFuture = new CompletableFuture<>();
        //添加链接监听
        bootstrap.connect(address).addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()) {
                //已经链接成功 为future设置结果  为当前channel
                channelFuture.complete(future.channel());
                log.info("rpc链接服务成功,地址:{}", address.toString());
                SocketUtil.bindAfterClient(future, workGroup);
            } else {
                future.channel().close().addListener(ChannelFutureListener.CLOSE);
                future.awaitUninterruptibly();
                workGroup.shutdownGracefully();
                throw new IllegalStateException("rpc链接服务失败,地址:" + address.toString());
            }
        });
        return channelFuture.get();
    }

    /**
     * 获取链接
     * @param address
     * @return
     */
    @SneakyThrows
    public Channel getChannel(InetSocketAddress address) {
        //从缓存中获取链接
        Channel channel = NettyClientContext.me.getChannel(address);
        if (channel == null) {
            //如果为空,创建客户端并设置缓存中
            channel = this.connected(address);
            NettyClientContext.me.setChannel(address, channel);
        }
        return channel;
    }


}



